package org.apache.lorry.zookeeper.meeting;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 参与者
 */
@Slf4j
public class Participator {
    private String name;
    private String ipPort;
    ZooKeeper zooKeeper;
    private volatile String speakerName;

    //默认状态为参与者
    private volatile MeetingState state = MeetingState.PARTICIPATOR;
    private volatile boolean connected = false;
    private volatile boolean expired = false;
    public final static int SESSION_TIMEOUT = 10000;
    private ThreadPoolExecutor executor;
    private String workStatus;

    protected MeetingCache assignedTasksCache = new MeetingCache();

    enum MeetingState {
        SPEAKER, PARTICIPATOR, LISTENER,
    }

    public Participator(String ipPort) {
        this.ipPort = ipPort;
        this.executor = new ThreadPoolExecutor(1, 1,
                1000L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(200));
    }

    public void init() throws IOException, InterruptedException, KeeperException {
        CountDownLatch waitSuccess = new CountDownLatch(1);
        ZooKeeper zk = new ZooKeeper(ipPort, SESSION_TIMEOUT, watchedEvent -> {
            if (watchedEvent.getType() == Watcher.Event.EventType.None) {
                switch (watchedEvent.getState()) {
                    case SyncConnected:
                        connected = true;
                        waitSuccess.countDown();
                        break;
                    case Disconnected:
                        connected = false;
                        break;
                    case Expired:
                        expired = true;
                        connected = false;
                        log.error("zookeeper session expired");
                        break;
                    default:
                        break;
                }
            } else
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                waitSuccess.countDown();
                log.info("zookeeper connected");
            }
        });
        try {
            waitSuccess.await(SESSION_TIMEOUT, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        this.zooKeeper = zk;
        initMeeting();
        name = getParticipantName(zooKeeper);
        log.info("new Participator name : {} in the meeting", name);
    }

    private void initMeeting() throws InterruptedException, KeeperException {
        Stat stat = ZkUtil.exists(ZkConstant.ROOT_PATH, zooKeeper);
        if (stat == null) {
            zooKeeper.create(ZkConstant.ROOT_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            log.info("create root path : {}", ZkConstant.ROOT_PATH);
        }
        Stat nameStat = ZkUtil.exists(ZkConstant.GET_PARTICIPANT_NAME_PATH, zooKeeper);
        if (nameStat == null) {
            zooKeeper.create(ZkConstant.GET_PARTICIPANT_NAME_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            log.info("create name path : {}", ZkConstant.GET_PARTICIPANT_NAME_PATH);
        }
        Stat  listenerStat = ZkUtil.exists(ZkConstant.LISTENER_PATH, zooKeeper);
        if (listenerStat == null) {
            zooKeeper.create(ZkConstant.LISTENER_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            log.info("create listener path : {}", ZkConstant.LISTENER_PATH);
        }
        Stat  statusStat = ZkUtil.exists(ZkConstant.STATUS_PATH, zooKeeper);
        if (statusStat == null) {
            zooKeeper.create(ZkConstant.STATUS_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            log.info("create status path : {}", ZkConstant.STATUS_PATH);
        }
        Stat  assignStat = ZkUtil.exists(ZkConstant.ASSIGN_PATH, zooKeeper);
        if (assignStat == null) {
            zooKeeper.create(ZkConstant.ASSIGN_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            log.info("create assign path : {}", ZkConstant.ASSIGN_PATH);
        }
    }

    public void becomeSpeaker() {
        log.info("{} begine to become speaker", name);
        zooKeeper.create(ZkConstant.SPEAKER_PATH, name.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,(int rc, String path, Object ctx, String name) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkSpeaker();     //如果连接失效，重新检查
                    break;
                case OK:
                    state = MeetingState.SPEAKER;
                    // TODO: 2023/11/30
                    log.info("{} become speaker success", this.name);
                    break;
                case NODEEXISTS:
                    log.info("speaker already exists , {} become listener", this.name);
                    state = MeetingState.LISTENER;
                    speakerExists();    //监听speaker节点

                    createAssignNode(); //创建assign节点

                    register(); //注册到listener节点

                    getTasks(); // 开始获取任务
                    break;
                default:
                    log.error("{} become speaker error", this.name, KeeperException.create(KeeperException.Code.get(rc), path));
            }
        },null);
    }

    void takeLeadership() {
        log.info("Going for list of workers");
    }

    void checkSpeaker() {
        zooKeeper.getData(ZkConstant.SPEAKER_PATH,false,(int rc, String path, Object ctx, byte[] data, Stat stat) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:    //如果连接失效，需要重新检查
                    checkSpeaker();
                    break;
                case OK:
                    if (new String(data).equals(name)) {    //如果是自己，成为speaker
                        log.info("{} : I am the speaker", name);
                        state = MeetingState.SPEAKER;
                        // TODO: 2023/11/30
                    } else {
                        log.info("{} : I am not the speaker, become a listener, and watch speaker", name);
                        state = MeetingState.LISTENER;
                        speakerExists();    //监听speaker节点
                    }
                    break;
                case NONODE:
                    log.info("speaker not exists");
                    becomeSpeaker();
                    break;
                default:
                    log.error("check speaker error", KeeperException.create(KeeperException.Code.get(rc), path));
            }
        },null);
    }

    void speakerExists() {
        zooKeeper.exists(ZkConstant.SPEAKER_PATH, e -> {
            if (e.getType() == Watcher.Event.EventType.NodeDeleted) {       //如果speaker节点被删除，重新选举
                assert ZkConstant.SPEAKER_PATH.equals(e.getPath());
                becomeSpeaker();
            }
        }, ((rc, path, ctx, stat) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    speakerExists();        //如果连接失效，重新检查
                    break;
                case OK:
                    log.info("speaker exists! watch it");
                    break;
                case NONODE:    //如果节点不存在，重新选举，状态回退到参与者
                    state = MeetingState.PARTICIPATOR;
                    log.info("speaker not exists now , so let's begin to elect a new speaker");
                    becomeSpeaker();
                    break;
                default:
                    checkSpeaker();
                    break;
            }
        }), null);
    }


    public String getParticipantName(ZooKeeper zooKeeper) {
        try {
            String namePath = zooKeeper.create(ZkConstant.GET_PARTICIPANT_NAME_PATH + "/No.", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            return namePath.replace(ZkConstant.GET_PARTICIPANT_NAME_PATH + "/", "");
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    private void createAssignNode() {
        zooKeeper.create(ZkConstant.ASSIGN_PATH+"/"+name, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,(int rc, String path, Object ctx, String name) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    createAssignNode();
                    break;
                case OK:
                    log.info("{} become listener success", this.name);
                    break;
                case NODEEXISTS:
                    log.info("{} already exists", this.name);
                    break;
                default:
                    log.error("{} become listener error", this.name, KeeperException.create(KeeperException.Code.get(rc), path));
            }
        },null);
    }

    private void register() {
        zooKeeper.create(ZkConstant.LISTENER_PATH+"/"+name, new byte[0],
                ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL,(int rc, String path, Object ctx, String name) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    register();
                    break;
                case OK:
                    log.info("{} Registered successfully", name);
                    break;
                case NODEEXISTS:
                    log.info("{} already exists", name);
                    break;
                default:
                    log.error("{} Something went wrong:", name, KeeperException.create(KeeperException.Code.get(rc), path));
            }
        },null);
    }

    private void getTasks() {
        zooKeeper.getChildren(ZkConstant.ASSIGN_PATH+"/"+name, e -> {
            if (e.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                assert (ZkConstant.ASSIGN_PATH+"/"+name).equals(e.getPath());
                getTasks();
            }
        }, (int rc, String path, Object ctx, java.util.List<String> children) -> {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    getTasks();
                    break;
                case OK:
                    if (children != null && !children.isEmpty()) {
                        log.info("I have {} tasks", children.size());
                        executor.execute(new Runnable() {   //防止回调线程阻塞
                            List<String> children;  //新添加的任务
                            AsyncCallback.DataCallback cb;

                            public Runnable init(List<String> children, AsyncCallback.DataCallback cb) {
                                this.children = children;
                                this.cb = cb;
                                return this;
                            }

                            @Override
                            public void run() {
                                if (children == null) { //
                                    return;
                                }
                                log.info("looping into tasks");
                                setWorkStatus("Working");
                                for (String task : children) {
                                    log.info("new task: {}", task);
                                    zooKeeper.getData(ZkConstant.ASSIGN_PATH + "/" + name + "/" + task, false, cb, task);
                                }
                            }
                        }.init(assignedTasksCache.addedAndSet(children), taskDataCallback));
                    }
                    break;
                default:
                    log.error("get tasks error", KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }, null);
    }

    AsyncCallback.DataCallback taskDataCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    zooKeeper.getData(path, false, taskDataCallback, null);
                    break;
                case OK:
                    executor.execute(new Runnable() {
                        byte[] data;
                        Object ctx;

                        public Runnable init(byte[] data, Object ctx) {
                            this.data = data;
                            this.ctx = ctx;
                            return this;
                        }

                        @Override
                        public void run() {
                            log.info("Executing your task: {}", new String(data));
                            zooKeeper.create(ZkConstant.STATUS_PATH + "/" + (String) ctx, "done".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, taskStatusCreateCallback, null);
                            zooKeeper.delete(ZkConstant.ASSIGN_PATH + "/" + name + "/" + (String) ctx, -1, (int rc, String path, Object rtx) -> {
                                switch (KeeperException.Code.get(rc)) {
                                    case CONNECTIONLOSS:
                                        break;
                                    case OK:
                                        log.info("Task correctly deleted: " + path);
                                        break;
                                    default:
                                        log.error("Failed to delete task data" + KeeperException.create(KeeperException.Code.get(rc), path));
                                }
                            }, null);
                        }
                    }.init(data, ctx));
                    break;
                default:
                    log.error("Failed to get task data: ", KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };


    AsyncCallback.StringCallback taskStatusCreateCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    zooKeeper.create(path, "done".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, taskStatusCreateCallback, null);
                    break;
                case OK:
                    log.info("Created status znode correctly: {}", name);
                    break;
                case NODEEXISTS:
                    log.warn("Node exists: {}", path);
                    break;
                default:
                    log.error("Something went wrong: ", KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    public void setWorkStatus(String status) {
        this.workStatus = status;   //本地保存workStatus
        updateWorkStatus(status);
    }

    private synchronized void updateWorkStatus(String workStatus) {
        if (workStatus == this.workStatus) {
            zooKeeper.setData(ZkConstant.LISTENER_PATH+"/"+name, workStatus.getBytes(), -1, (int rc, String path, Object ctx, Stat stat) -> {
                switch (KeeperException.Code.get(rc)) {
                    case CONNECTIONLOSS:
                        updateWorkStatus((String) ctx);
                }
            },workStatus);
        }
    }


    public String name() {
        return name;
    }

}